package com.atguigu;

import beans.LoginEvent;
import beans.LoginFailWarning;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

/**
 * @author zkq
 * @date 2022/10/15 11:54
 */
public class Login_Fail_Detect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //从文件中读取数据
        URL resource = Login_Fail_Detect.class.getResource("/LoginLog.csv");
        String path = resource.getPath();
        SingleOutputStreamOperator<LoginEvent> LoginStream = env.readTextFile(path)
                .map(data -> {
                    String[] split = data.split(",");
                    return new LoginEvent(new Long(split[0]), split[1], split[2], Long.valueOf(split[3]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                            @Override
                            public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        })
                );
        //按用户id分流，第一次遇到登陆失败设置定时器，2秒后触发，触发时检测状态中存储的登录失败的个数，如果大于等于2次就输出报警信息
        SingleOutputStreamOperator<LoginFailWarning> result = LoginStream
                .keyBy(LoginEvent::getUserId)
                .process(new LoginFailDetectWarning(2));

        result.print();
        env.execute("login fail detect");
    }
    public static class LoginFailDetectWarning extends KeyedProcessFunction<Long,LoginEvent,LoginFailWarning>{
        //定义属性，最大连续登录失败次数
        private Integer maxFailTimes;
        private ListState<LoginEvent> listState;
        private ValueState<Long> state;

        public LoginFailDetectWarning(Integer maxFailTimes) {
            this.maxFailTimes = maxFailTimes;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
             //定义状态，保存两秒内所有的登陆失败事件
            listState = getRuntimeContext().getListState(new ListStateDescriptor<LoginEvent>("login-fail-list", LoginEvent.class));
            //定义状态：保存注册的定时器时间戳，因为中间如果有成功需要删除定时器
            state = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts", Long.class));
        }

        @Override
        public void processElement(LoginEvent value, Context ctx, Collector<LoginFailWarning> out) throws Exception {
            //判断当前登陆时间类型
            if ("fail".equals(value.getLoginState())){
                //如果是失败事件，添加到列表状态中
                listState.add(value);
                //如果没有定时器，注册一个两秒之后的定时器
                if (state.value() == null){
                    Long ts = (value.getTimestamp() + 2)*1000L;
                    ctx.timerService().registerEventTimeTimer(ts);
                    state.update(ts);
                }
            }else {
                //如果登陆成功，删除定时器，清空状态，重新开始
                if (state.value() != null){
                    ctx.timerService().deleteEventTimeTimer(state.value());
                }
                listState.clear();
                state.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailWarning> out) throws Exception {
            //定时器触发，说明2秒内无登录成功事件，判断是否大于2次，报警
            ArrayList<LoginEvent> loginFailEvents = new ArrayList<>();
            for (LoginEvent loginEvent : listState.get()) {
                loginFailEvents.add(loginEvent);
            }
            Integer failTimes = loginFailEvents.size();

            if(failTimes >= maxFailTimes){
                //如果超出设定的最大失败次数，输出报警
                out.collect(new LoginFailWarning(ctx.getCurrentKey(),loginFailEvents.get(0).getTimestamp(),
                        loginFailEvents.get(failTimes-1).getTimestamp(),"login fail in 2s for "+failTimes+" times"));
            }
            //清空状态
            listState.clear();
            state.clear();
        }
        //改进     可以不用定时器 来一条登录就判断一下 符合连续两次立刻报警 不给坏人可乘之机 两秒有可能就破开密码了
        /*@Override
        public void processElement(LoginEvent value, Context ctx, Collector<LoginFailWarning> out) throws Exception {
            if ("fail".equals(value.getLoginState())) {
                Iterator<LoginEvent> iterator = loginFailListState .get().iterator();
                if( iterator.hasNext() ){
                    LoginEvent firstFail = iterator.next();
// 如果两次失败时间间隔小于2 秒，输出报警
                    if( value.getTimestamp() - firstFail.getTimestamp() <= 2 ){ out.collect( new LoginFailWarning(value.getUserId(),
                            firstFail.getTimestamp(),
                            value.getTimestamp(),
                            "login fail in 2s") );



                    }
                    loginFailListState .clear();
                    //比上面更严谨了 清理完把这次失败再添加进状态
                    loginFailListState .add(value);
                }else {
                    loginFailListState .add(value);
                }
            }else
                loginFailListState .clear();
        }*/

    }
}
